package org.huangrui.spark.java.core.rdd.operate.transform;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * @Author hr
 * @Create 2024-10-17 15:49
 */
public class Spark09_Operate_Transform_KV_groupBy {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);

        final List<Integer> nums = Arrays.asList(1, 2, 3, 4);
        final JavaRDD<Integer> rdd = jsc.parallelize(nums, 2);

        // TODO groupBy：按照指定的规则对数据进行分组
        //           给每一个数据增加一个标记，相同的标记的数据会放置在一个组中，这个标记就是组名
        //      groupBy结果：就是KV类型的数据
        //           (0，【2，4】) => (0，6)
        //           (1, 【1，3】) => (1，4)
        /*
         1 => 1
         2 => 0
         3 => 1
         4 => 0
         */
        JavaPairRDD<Integer, Integer> mapValues = rdd.groupBy(x -> x % 2).mapValues(v1 -> {
            Iterator<Integer> iter = v1.iterator();
            int sum = 0;
            while (iter.hasNext()) {
                sum += iter.next();
            }
            return sum;
        });
        mapValues.collect().forEach(System.out::println);
        jsc.close();
    }
}
